liburingutils: Add API to handle socket message asynchronous Bug: 385143770 Test: Unit test Change-Id: I32ed7511e4e94e43766acb9ef3ddffc5d2eb0320 Signed-off-by: Akilesh Kailash <akailash@google.com>
diff --git a/Android.bp b/Android.bp index 5fa331e..f625107 100644 --- a/Android.bp +++ b/Android.bp
@@ -6,15 +6,19 @@ name: "liburingutils", srcs: [ - "src/LibUringUtils.cpp", + "src/IOUringSocketHandler.cpp", ], cflags: ["-Werror"], export_include_dirs: ["include"], + static_libs: [ + "liburing", + ], shared_libs: [ "libbase", + "liblog", ], tidy: true, @@ -32,10 +36,13 @@ } cc_test { - name: "liburingutils_tests", + name: "IOUringSocketHandler_tests", test_suites: ["device-tests"], srcs: [ - "src/LibUringUtils_test.cpp", + "src/IOUringSocketHandler_test.cpp", + ], + static_libs: [ + "liburing", ], shared_libs: [ "liburingutils", diff --git a/include/IOUringSocketHandler/IOUringSocketHandler.h b/include/IOUringSocketHandler/IOUringSocketHandler.h new file mode 100644 index 0000000..226a5b2 --- /dev/null +++ b/include/IOUringSocketHandler/IOUringSocketHandler.h
@@ -0,0 +1,174 @@ +/* + * Copyright (C) 2025 The Android Open Source Project + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#pragma once + +#include <memory> +#include <vector> + +#include <liburing.h> + +/* + * IOUringSocketHandler is a helper class for using io_uring with a socket. + * + * Typical usage from a given thread: + * + * As a one time setup: + * 1. Create an instance of IOUringSocketHandler with the socket file descriptor. + * 2. Setup io_uring ring buffer. + * 3. Allocate buffers for the ring buffer. + * 4. Register buffers with io_uring. + * 5. EnqueueMultishotRecvmsg() will submit the SQE to receive the data + * + * In the I/O path: + * + * 6. Receive data from the socket through ReceiveData() + * 7. Release the buffer to io_uring. + * + * Note that the thread which sets up the io_uring instance should handle the + * I/O through ReceiveData() call. + */ + +class IOUringSocketHandler { +public: + IOUringSocketHandler(int socket_fd); + ~IOUringSocketHandler(); + + // Setup io_uring ring buffer + // queue_size: The size of the io_uring submission queue. + // Determines the maximum number of outstanding I/O requests. + // return: true on success, false on failure (e.g., if io_uring_setup fails). + // + // This function initializes the io_uring context and sets up the submission + // and completion queues. It prepares the io_uring instance for I/O operations. + // A larger queue_size allows for more concurrent I/O operations but consumes + // more memory. + bool SetupIoUring(int queue_size); + + // Allocate 'num_buffers' of size 'buf_size' + // + // num_buffers: The number of buffers to allocate. + // buf_size: The size of each buffer in bytes. + // + // This function allocates a set of buffers that will be used for I/O operations + // with io_uring. These buffers are typically used to hold data that is read from + // or written to files or sockets. The allocated buffers are managed internally + // and are later registered with io_uring. + // + // The num_buffers will be the payload for the caller. Internally, it + // allocates additional metadata: + // a: sizeof(struct ucred) + sizeof(struct cmsghdr) + // b: sizeof(struct io_uring_recvmsg_out) + // This allows sender to send the ucred credential information if required. + // + // This function also registers the allocated buffers with the io_uring instance. + // Registering buffers allows the kernel to access them directly, avoiding the need + // to copy data between user space and kernel space during I/O operations. This + // improves performance. + // + // Please see additional details on how num_buffers will be used + // by the io_uring: https://man7.org/linux/man-pages/man3/io_uring_setup_buf_ring.3.html + bool AllocateAndRegisterBuffers(size_t num_buffers, size_t buf_size); + + // Free up registered buffers with the io_uring instance. + // + // All the buffers allocated using AllocateAndRegisterBuffers() API will be + // freed and de-registered. Callers can then call + // AllocateAndRegisterBuffers() to re-register new set of bufferes with the + // ring. + void DeRegisterBuffers(); + + // ARM io_uring recvmsg opcode + // + // return: true on success, false on failure (e.g., if submission queue is full). + // + // This function enqueues a "multishot recvmsg" operation into the io_uring submission queue. + // Multishot recvmsg allows receiving multiple messages from a socket with a single + // io_uring submission. The function prepares the submission queue + // entry (SQE) for the recvmsg operation. + bool EnqueueMultishotRecvmsg(); + + // Release the buffer to io_uring + // + // This function releases a buffer back to the io_uring subsystem after it has been + // used for an I/O operation. This makes the buffer available for reuse in subsequent + // I/O operations. + // + // Additionally, when the buffer is released, a check is done to see if + // there are more CQE entries available. If not, EnqueueMultishotRecvmsg() + // is invoked so that the SQE submission is done for receiving next set of + // I/O. + void ReleaseBuffer(); + + // Receive payload data of size payload_len. Additionally, receive + // credential data. + // + // payload: A pointer to a void pointer. This will be set to point to the received + // payload data. + // + // payload_len: A reference to a size_t. This will be set to the length of the + // received payload data. + // + // cred: A pointer to a struct ucred pointer. This will be set to point to the + // user credentials associated with the received data (if available). + // If the sender doesn't have credential information in the payload, + // then nullptr will be returned. + // + // This function retrieves the data received from a recvmsg operation. It extracts the payload + // data and its length, as well as the user credentials associated with the sender. The + // caller is responsible for freeing the allocated memory for the payload and credentials + // when they are no longer needed. + void ReceiveData(void** payload, size_t& payload_len, struct ucred** cred); + + // check if io_uring is supported + // + // return: true if io_uring is supported by the kernel, false otherwise. + // + // This function checks if the io_uring feature is supported by the underlying Linux kernel. + static bool isIouringEnabled(); + +private: + static bool isIouringSupportedByKernel(); + // Register buffers with io_uring + // + // return: true on success, false on failure (e.g., if io_uring_register_buffers fails). + // + // This function registers the previously allocated buffers with the io_uring instance. + // Registering buffers allows the kernel to access them directly, avoiding the need + // to copy data between user space and kernel space during I/O operations. This + // improves performance. + bool RegisterBuffers(); + + struct uring_context { + struct io_uring ring; + }; + // Socket fd + int socket_; + std::unique_ptr<uring_context> mCtx; + std::vector<std::unique_ptr<uint8_t[]>> buffers_; + struct msghdr msg; + int control_len_; + size_t num_buffers_ = 0; + int buffer_size_; + int active_buffer_id_ = -1; + struct io_uring_cqe* cqe; + // A constant buffer group id as we don't support multiple buffer groups + // yet. + const int bgid_ = 7; + struct io_uring_buf_ring* br_; + bool registered_buffers_ = false; + bool ring_setup_ = false; +}; diff --git a/include/liburingutils/LibUringUtils.h b/include/liburingutils/LibUringUtils.h deleted file mode 100644 index a6dbefc..0000000 --- a/include/liburingutils/LibUringUtils.h +++ /dev/null
@@ -1,28 +0,0 @@ -/* - * Copyright (C) 2025 The Android Open Source Project - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#ifndef __LIBURING_UTILS_H -#define __LIBURING_UTILS_H - -class LibUringUtils { -public: - static bool isIouringEnabled(); - -private: - static bool isIouringSupportedByKernel(); -}; - -#endif diff --git a/src/IOUringSocketHandler.cpp b/src/IOUringSocketHandler.cpp new file mode 100644 index 0000000..0c68281 --- /dev/null +++ b/src/IOUringSocketHandler.cpp
@@ -0,0 +1,206 @@ +/* + * Copyright (C) 2025 The Android Open Source Project + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#define LOG_TAG "IOUringSocketHandler" + +#include <sys/resource.h> +#include <sys/utsname.h> +#include <unistd.h> + +#include <limits.h> +#include <linux/time_types.h> +#include <sys/cdefs.h> +#include <sys/prctl.h> +#include <sys/socket.h> +#include <sys/types.h> +#include <sys/un.h> +#include <unistd.h> + +#include <chrono> +#include <thread> + +#include <cutils/sockets.h> +#include <private/android_logger.h> + +#include <IOUringSocketHandler/IOUringSocketHandler.h> + +#include <android-base/logging.h> +#include <android-base/scopeguard.h> + +bool IOUringSocketHandler::isIouringEnabled() { + return isIouringSupportedByKernel(); +} + +bool IOUringSocketHandler::isIouringSupportedByKernel() { + struct utsname uts {}; + unsigned int major, minor; + + uname(&uts); + if (sscanf(uts.release, "%u.%u", &major, &minor) != 2) { + return false; + } + + // We will only support kernels from 6.1 and higher. + return major > 6 || (major == 6 && minor >= 1); +} + +IOUringSocketHandler::IOUringSocketHandler(int socket_fd) : socket_(socket_fd) {} + +IOUringSocketHandler::~IOUringSocketHandler() { + DeRegisterBuffers(); + if (ring_setup_) { + io_uring_queue_exit(&mCtx->ring); + } +} + +bool IOUringSocketHandler::EnqueueMultishotRecvmsg() { + struct io_uring_sqe* sqe = io_uring_get_sqe(&mCtx->ring); + memset(&msg, 0, sizeof(msg)); + msg.msg_controllen = control_len_; + io_uring_prep_recvmsg_multishot(sqe, socket_, &msg, 0); + sqe->flags |= IOSQE_BUFFER_SELECT; + sqe->buf_group = bgid_; + int ret = io_uring_submit(&mCtx->ring); + if (ret < 0) { + LOG(ERROR) << "EnqueueMultishotRecvmsg failed: ret: " << ret; + return false; + } + return true; +} + +bool IOUringSocketHandler::AllocateAndRegisterBuffers(size_t num_buffers, size_t buf_size) { + num_buffers_ = num_buffers; + control_len_ = CMSG_ALIGN(sizeof(struct ucred)) + sizeof(struct cmsghdr); + + buffer_size_ = sizeof(struct io_uring_recvmsg_out) + control_len_ + buf_size; + + for (size_t i = 0; i < num_buffers_; i++) { + std::unique_ptr<uint8_t[]> buffer = std::make_unique<uint8_t[]>(buffer_size_); + buffers_.push_back(std::move(buffer)); + } + return RegisterBuffers(); +} + +bool IOUringSocketHandler::RegisterBuffers() { + int ret = 0; + br_ = io_uring_setup_buf_ring(&mCtx->ring, num_buffers_, bgid_, 0, &ret); + if (!br_) { + LOG(ERROR) << "io_uring_setup_buf_ring failed with error: " << ret; + return false; + } + for (size_t i = 0; i < num_buffers_; i++) { + void* buffer = buffers_[i].get(); + io_uring_buf_ring_add(br_, buffer, buffer_size_, i, io_uring_buf_ring_mask(num_buffers_), + i); + } + io_uring_buf_ring_advance(br_, num_buffers_); + LOG(DEBUG) << "RegisterBuffers success: " << num_buffers_; + registered_buffers_ = true; + return true; +} + +void IOUringSocketHandler::DeRegisterBuffers() { + if (registered_buffers_) { + io_uring_free_buf_ring(&mCtx->ring, br_, num_buffers_, bgid_); + } + buffers_.clear(); + num_buffers_ = 0; + control_len_ = 0; + buffer_size_ = 0; +} + +bool IOUringSocketHandler::SetupIoUring(int queue_size) { + mCtx = std::unique_ptr<uring_context>(new uring_context()); + struct io_uring_params params = {}; + + // COOP_TASKRUN - No IPI to logd + // SINGLE_ISSUER - Only one thread is doing the work on the ring + // TASKRUN_FLAG - we use peek_cqe - Hence, trigger task work if required + // DEFER_TASKRUN - trigger task work when CQE is explicitly polled + params.flags |= (IORING_SETUP_COOP_TASKRUN | IORING_SETUP_SINGLE_ISSUER | + IORING_SETUP_TASKRUN_FLAG | IORING_SETUP_DEFER_TASKRUN); + + int ret = io_uring_queue_init_params(queue_size + 1, &mCtx->ring, ¶ms); + if (ret) { + LOG(ERROR) << "io_uring_queue_init_params failed with ret: " << ret; + return false; + } else { + LOG(INFO) << "io_uring_queue_init_params success"; + } + + ring_setup_ = true; + return true; +} + +void IOUringSocketHandler::ReleaseBuffer() { + if (active_buffer_id_ == -1) { + return; + } + + // Put the buffer back to the pool + io_uring_buf_ring_add(br_, buffers_[active_buffer_id_].get(), buffer_size_, active_buffer_id_, + io_uring_buf_ring_mask(num_buffers_), 0); + io_uring_buf_ring_cq_advance(&mCtx->ring, br_, 1); + active_buffer_id_ = -1; + + // If there are no more CQE data, re-arm the SQE + bool is_more_cqe = (cqe->flags & IORING_CQE_F_MORE); + if (!is_more_cqe) { + EnqueueMultishotRecvmsg(); + } +} + +void IOUringSocketHandler::ReceiveData(void** payload, size_t& payload_len, struct ucred** cred) { + if (io_uring_peek_cqe(&mCtx->ring, &cqe) < 0) { + int ret = io_uring_wait_cqe(&mCtx->ring, &cqe); + if (ret) { + LOG(ERROR) << "WaitCqe failed: " << ret; + EnqueueMultishotRecvmsg(); + return; + } + } + + if (cqe->res < 0) { + io_uring_cqe_seen(&mCtx->ring, cqe); + EnqueueMultishotRecvmsg(); + return; + } + + active_buffer_id_ = cqe->flags >> IORING_CQE_BUFFER_SHIFT; + + void* this_recv = buffers_[active_buffer_id_].get(); + struct io_uring_recvmsg_out* o = io_uring_recvmsg_validate(this_recv, cqe->res, &msg); + + if (!o) { + return; + } + + struct cmsghdr* cmsg; + cmsg = io_uring_recvmsg_cmsg_firsthdr(o, &msg); + + struct ucred* cr = nullptr; + while (cmsg != nullptr) { + if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_CREDENTIALS) { + cr = (struct ucred*)CMSG_DATA(cmsg); + break; + } + cmsg = io_uring_recvmsg_cmsg_nexthdr(o, &msg, cmsg); + } + + *payload = io_uring_recvmsg_payload(o, &msg); + payload_len = io_uring_recvmsg_payload_length(o, cqe->res, &msg); + *cred = cr; +} diff --git a/src/LibUringUtils_test.cpp b/src/IOUringSocketHandler_test.cpp similarity index 62% rename from src/LibUringUtils_test.cpp rename to src/IOUringSocketHandler_test.cpp index aff5206..67361f6 100644 --- a/src/LibUringUtils_test.cpp +++ b/src/IOUringSocketHandler_test.cpp
@@ -14,20 +14,18 @@ * limitations under the License. */ -#include <liburingutils/LibUringUtils.h> +#include <IOUringSocketHandler/IOUringSocketHandler.h> #include <gtest/gtest.h> -class LibUringUtilsTest : public testing::Test { +class IOUringSocketHandlerTest : public testing::Test { public: void testIsIouringEnabled(bool expectedResult) { - EXPECT_EQ(LibUringUtils::isIouringEnabled(), expectedResult); + EXPECT_EQ(IOUringSocketHandler::isIouringEnabled(), expectedResult); } }; -TEST_F(LibUringUtilsTest, ReturnsIouringNotEnabled) { - // TODO: b/385143770 - Change this test to base on the real OS version, - // this default expected value is true for the binary built from the - // latest version of source code. - testIsIouringEnabled(true); +TEST_F(IOUringSocketHandlerTest, ReturnsIouringNotEnabled) { + // TODO: b/385143770 - Change this behavior to check the OS version and Liburing version. + testIsIouringEnabled(false); } diff --git a/src/LibUringUtils.cpp b/src/LibUringUtils.cpp deleted file mode 100644 index 25151d3..0000000 --- a/src/LibUringUtils.cpp +++ /dev/null
@@ -1,41 +0,0 @@ -/* - * Copyright (C) 2025 The Android Open Source Project - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -#define LOG_TAG "LibUringUtils" - -#include <android-base/strings.h> -#include <liburingutils/LibUringUtils.h> -#include <sys/resource.h> -#include <sys/utsname.h> -#include <unistd.h> - -bool LibUringUtils::isIouringEnabled() { - // TODO: b/385143770 - Change this behavior to also check the Liburing version. - return isIouringSupportedByKernel(); -} - -bool LibUringUtils::isIouringSupportedByKernel() { - struct utsname uts {}; - unsigned int major, minor; - - uname(&uts); - if (sscanf(uts.release, "%u.%u", &major, &minor) != 2) { - return false; - } - - // We will only support kernels from 6.1 and higher. - return major > 6 || (major == 6 && minor >= 1); -} 